package zhuzx.scheduler;

import java.util.Date;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 任务调用器，通过唯一的构造方法传入具体的任务实体，按需调用。
 * @author	zhuzx	2016年1月8日
 */
public class MultitaskInvoker {
	private static final int THREAD_NUM = 6;
	
	private TaskContainer taskContainer;
	private ExecutorService pool;
	private BlockingQueue<Future<?>> futuresBuffer;
	private Thread submitter;
	private Thread cleaner;
	private Thread actuator;

	public MultitaskInvoker(TaskContainer taskContainer) {
		this.taskContainer = taskContainer;
	}
	
	/**
	 * 以特定频率串行执行任务，按给定的时间启动和停止
	 * @param period 连续启动两次任务的间隔，固定执行频率
	 * @param startTime 启动时间
	 * @param endTime 停止时间
	 */
	public void startTaskSerial(long period, Date startTime, Date endTime) {
		/* 若scheduleAtFixedRate方法执行任务的时间超过了设定的时间间隔，则会等待当前任务完成之后马上执行下一次任务，而不会并发执行。
		 * 用newScheduledThreadPool方法创建的多个线程的线程池无法并发合作执行，
		 * 比如第一个任务已经超时了，第二个任务还是要等待第一个任务结束后才开始，其它空闲线程就在那里傻看着，
		 * 而不是在设定的delayMillis到达之后让其它空闲线程如期调用第二个任务，这个线程池是用来启动多批次的任务调度时用的。
		 * 观察到控制台输出的频率时快时慢，这并非是由于有其它线程在并行执行，创建单个线程的池也会如此，这是由于任务本身不能稳定地完成。
		 * 用newSingleThreadScheduledExecutor方法创建的单个线程的线程池，在该线程执行任务失败后会再创建一个新线程取而代之，
		 * 以继续执行后续任务，这是与newScheduledThreadPool(1)的区别，后者的这个线程死了就完了。
		 */
		ScheduledExecutorService ses = Executors.newScheduledThreadPool((endTime == null) ? 1 : 2);
		pool = ses;
		long curr = System.currentTimeMillis();
		long initialDelay = (startTime == null) ? 0 : startTime.getTime() - curr;
		ses.scheduleAtFixedRate(taskContainer, initialDelay, period, TimeUnit.MILLISECONDS);
		if (endTime != null) {
			long timeLeft = endTime.getTime() - curr;
			timeLeft = (timeLeft < 0) ? 0 : timeLeft;
			ses.schedule(
					new Runnable() {
						public void run() {
							shutdown();
						}
					}, timeLeft, TimeUnit.MILLISECONDS);
		}
	}
	
	/**
	 * 以特定频率串行执行任务，需要手动停止
	 * @param period 连续启动两次任务的间隔，固定执行频率
	 */
	public void startTaskSerial(long period) {
		ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor();
		pool = ses;
		ses.scheduleAtFixedRate(taskContainer, 0, period, TimeUnit.MILLISECONDS);
	}
	
	/**
	 * 并行执行任务，阻塞式
	 * @param waitingMillis 单个任务的最大等待时间
	 */
	public void startTaskParallelBlocking(long waitingMillis) {
		pool = Executors.newFixedThreadPool(THREAD_NUM);
		futuresBuffer = new ArrayBlockingQueue<Future<?>>(THREAD_NUM * 20);
		submitter = new Thread(new Submitter());
		submitter.start();
		cleaner = new Thread(new Cleaner(waitingMillis));
		cleaner.start();
	}
	
	/**
	 * 并行执行任务，自检式
	 * @param checkingPeriod 自检的周期
	 */
	public void startTaskParallelRechecking(long checkingPeriod) {
		pool = Executors.newFixedThreadPool(THREAD_NUM);
		actuator = new Thread(new SelfCheckActuator(checkingPeriod));
		actuator.start();
	}
	
	/**
	 * 统一的关闭方法
	 */
	public void shutdown() {
		if (submitter != null) {
			submitter.interrupt();
			cleaner.interrupt();
		}
		if (actuator != null) {
			actuator.interrupt();
		}
		pool.shutdown();
		System.out.println(">>关闭线程池...");
	}
	
	/**
	 * 清理阻塞队列中已完成的任务
	 */
	public class Cleaner implements Runnable {
		private long waitingMillis;

		public Cleaner(long waitingMillis) {
			this.waitingMillis = waitingMillis;
		}

		public void run() {
			long cancelCounter = 0;
			try {
				Future<?> headFu = null;
				while (!Thread.currentThread().isInterrupted()) {
					headFu = (headFu != null) ? headFu : futuresBuffer.take();
					if (headFu.isDone()) {
						headFu = futuresBuffer.take();
					} else {
						if (futuresBuffer.size()>0 && futuresBuffer.peek().isDone()) {
							futuresBuffer.take();
						} else {
							try {
								headFu.get(waitingMillis, TimeUnit.MILLISECONDS);
							} catch (TimeoutException te) {
								System.out.println(">>>>任务超时");
								headFu.cancel(true);
								cancelCounter++;
							}
						}
					}
				}
				System.out.println(">>清理线程中断-正常");
			} catch (InterruptedException e) {
				System.out.println(">>清理线程中断-阻塞");
				Thread.currentThread().interrupt();
			} catch (ExecutionException e) {
				e.printStackTrace();
			} finally {
				System.out.println("取消任务数：" + cancelCounter);
			}
		}
	}
	
	/**
	 * 往阻塞队列中添加（提交）任务
	 */
	public class Submitter implements Runnable {
		public void run() {
			long counter = 0;
			try {
				while (!Thread.currentThread().isInterrupted()) {
					futuresBuffer.put(pool.submit(taskContainer));
					counter++;
				}
				System.out.println(">>提交线程中断-正常");
			} catch (InterruptedException e) {
				System.out.println(">>提交线程中断-阻塞");
				Thread.currentThread().interrupt();
			} finally {
				System.out.println("提交任务数：" + counter);
			}
		}
	}
	
	/**
	 * 自检式的执行器，提交和清理都由本身完成，按照给定的checkingPeriod周期性自检
	 */
	public class SelfCheckActuator implements Runnable {
		private long checkingPeriod;
		
		public SelfCheckActuator(long checkingPeriod) {
			this.checkingPeriod = checkingPeriod;
		}

		public void run() {
			long counter = 0;
			LinkedHashSet<Future<?>> futures = new LinkedHashSet<Future<?>>(THREAD_NUM);
			while (!Thread.currentThread().isInterrupted()) {
				if (futures.size() < THREAD_NUM) {
					futures.add(pool.submit(taskContainer));
					counter++;
				} else {
					Iterator<Future<?>> it = futures.iterator();
					while (it.hasNext()) {
						if (it.next().isDone()) {
							it.remove();
						}
					}
					try {
						Thread.sleep(checkingPeriod);
					} catch (InterruptedException e) {
						Thread.currentThread().interrupt();
					}
				}
			}
			System.out.println("提交任务数：" + counter);
		}
	}
	
}
